package edu.npu.GraphQuery;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Scanner;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Mapper;

import edu.npu.GraphIndex.BasicGraphBFS;
import edu.npu.GraphIndex.CntVid;
import edu.npu.GraphIndex.HistogramInit;
import edu.npu.GraphQuery.SPGraphQuery.QEdge;

public class GraphQueryDriver {
	public static int partNum = 3;
	public static int HOP = 2;
	public static List<List<QEdge>> queryorder = new ArrayList<List<QEdge>>();
	public static final String INDEX_FILE_PATH = "/home/zhao/index";
	public static final String QUERY_GRAPH_PATH = "hdfs://test118:9000/home/zhao/query/";
	public static Map<Long, Integer> distance = new HashMap<Long, Integer>();
	public static Map<Long, Map<Integer, Map<String, CntVid>>> histogramIndex = 
		new HashMap<Long, Map<Integer, Map<String, CntVid>>>();
	public static List<Map<Long, Map<Integer, Map<String, CntVid>>>> qHistogramIndexes = 
		new ArrayList<Map<Long, Map<Integer,Map<String,CntVid>>>>();
	public static Map<String, List<Long>> lookupTable = new HashMap<String, List<Long>>();
	public static List<Map<Long, String>> qLabels = new ArrayList<Map<Long, String>>();
	public static List<Map<Long, List<Long>>> qGraphs = new ArrayList<Map<Long,List<Long>>>();
	public static List<List<PartGraph>> results = new ArrayList<List<PartGraph>>();
	public static Set<QEdge> boundaryEdge = new HashSet<QEdge>();
	
	//在静态块里实现全局变量，使其加入内存,并且保证每个节点做一次，
	//如果放在map中一个节点做了多次
	static {
		Scanner scanner = new Scanner(INDEX_FILE_PATH);
		String str = null;
		//重新构建distance
		while (scanner.hasNext()) {
			str = scanner.next();
			if(str.startsWith("#lookuptable"))
				break;
			int index = str.indexOf("\t");
			distance.put(Long.parseLong(str.substring(0, index)), 
					Integer.parseInt(str.substring(index + 1)));
		}
		//重新构建lookupTable
		while(scanner.hasNext()) {
			str = scanner.next();
			if(str.startsWith("#histogram"))
				break;
			String key = str.substring(0, str.indexOf("\t"));
			if(!lookupTable.containsKey(key))
				lookupTable.put(key, new ArrayList<Long>());
			String[] nodes = str.substring(str.indexOf("\t") + 1).split("#");
			for(String node : nodes) {
				lookupTable.get(key).add(Long.parseLong(node));
			}
		}
		//重新构建histogram
		while(scanner.hasNext()) {
			str = scanner.next();
			String[] splits = str.split("\t");
			long key = Long.parseLong(splits[0]);
			int dist = Integer.parseInt(splits[1]);
			if(!histogramIndex.containsKey(key))
				histogramIndex.put(key, new HashMap<Integer, Map<String, CntVid>>());
			if(!histogramIndex.get(key).containsKey(dist))
				histogramIndex.get(key).put(dist, new HashMap<String, CntVid>());
			CntVid cv = new CntVid();
			for(int i = 3; i < splits.length; i++)
				cv.add(Long.parseLong(splits[i]));
			histogramIndex.get(key).get(dist).put(splits[2], cv);
		}
		scanner.close();
		//重新构建boundary
		
		//读取查询图
		Configuration conf = new Configuration();
		FileSystem fs; 
		long node;
		try {
			fs = FileSystem.get(conf);
			for(int num = 1; num <= partNum; num++) {
				Map<Long, List<Long>> qGraph = new HashMap<Long, List<Long>>();
				scanner = new Scanner(fs.open(new Path(QUERY_GRAPH_PATH + "qGraph + /" + num)));
				
				while(scanner.hasNext()) {
					node = scanner.nextLong();
					if(!qGraph.containsKey(node))
						qGraph.put(node, new ArrayList<Long>());
					qGraph.get(node).add(scanner.nextLong());
				}
				scanner.close();
				scanner = new Scanner(fs.open(new Path(QUERY_GRAPH_PATH + "qLabel + /" + num)));
				Map<Long, String> qLabel = new HashMap<Long, String>();
				while(scanner.hasNext()) {
					node = scanner.nextLong();
					qLabel.put(node, scanner.next());
				}
				Map<Long, Map<Integer, Map<String, CntVid>>> qHistogramIndex = 
					new HashMap<Long, Map<Integer,Map<String,CntVid>>>();
				BasicGraphBFS qHistBgb = new BasicGraphBFS();
				HistogramInit qHistInit = new HistogramInit();
				for(Entry<Long, String> entry : qLabel.entrySet()) {
					if(!qHistogramIndex.containsKey(entry.getKey())) {
						qHistogramIndex.put(entry.getKey(), new HashMap<Integer, Map<String, CntVid>>());
					}
					qHistInit.histogram = qHistogramIndex.get(entry.getKey());
					//建立遍历队列
					for(long n : qGraph.get(entry.getKey())) {
						qHistInit.queue.add(n);
					}
					qHistBgb.queue = qHistInit.queue;
					qHistInit.label = qLabel;
					qHistInit.traversed.add(entry.getKey());
					qHistBgb.traverse(qGraph, qHistInit);
					qHistInit.clear();
				}
				SPGraphQuery spGQuery = new SPGraphQuery(qGraph, 
						lookupTable, qLabel, histogramIndex, qHistogramIndex);
				List<Map<QEdge, QEdge>> result = spGQuery.queryProcessing();
				List<PartGraph> partGraphs = new ArrayList<PartGraph>();
				for(Map<QEdge, QEdge> g : result) {
					partGraphs.add(new PartGraph(g, 1 << num - 1));
				}
				results.add(partGraphs);
			}
			scanner.close();
			//内部join
			for (int j = 0; j < partNum - 1; j++) {
				innerJoin(j);
			}
			//如果有匹配成功的在最后一行里
			for(PartGraph pg : results.get(partNum - 1)) {
				//TODO 匹配成功的子图输出
				if(pg.joinNum == (Math.pow(2, partNum + 1) - 1)) {
					System.out.println(pg.toString());
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	public static class GraphQueryMapper extends Mapper<LongWritable, LongWritable, LongWritable, LongWritable> {
		@Override
		protected void map(LongWritable key, LongWritable value, Context context)
				throws IOException, InterruptedException {
			
		}
	}
	
	//判断对于查询图所有能够join的边（相同边）原图是否也全部能够join
	public static boolean isJoin(PartGraph graph1, PartGraph graph2) {
		if(graph1.equals(graph2))
			return false;
		Set<QEdge> edges = graph1.graph.keySet();
		edges.retainAll(graph2.graph.keySet());
		for(QEdge qe : edges) {
			if(!graph1.graph.get(qe).equals(graph2.graph.get(qe))) {
				return false;
			}
		}
		return true;
	}
	public static void innerJoin(int index) {
		for (int i = index + 1; i < partNum; i++) {
			for (PartGraph g1 : results.get(index)) {
				for (PartGraph g2 : results.get(i)) {
					if (isJoin(g1, g2)) {
						PartGraph pg = new PartGraph();
						pg.joinNum = g1.joinNum | g2.joinNum;
						pg.graph = new HashMap<SPGraphQuery.QEdge, SPGraphQuery.QEdge>();
						for (Entry<QEdge, QEdge> entry : g1.graph.entrySet()) {
							if (!pg.graph.containsKey(entry.getKey()))
								pg.graph.put(entry.getKey(), entry.getValue());
						}
						for (Entry<QEdge, QEdge> entry : g2.graph.entrySet()) {
							if (!pg.graph.containsKey(entry.getKey()))
								pg.graph.put(entry.getKey(), entry.getValue());
						}
						results.get(index + 1).add(pg);
					}
				}
			}
		}
	}
}
class PartGraph implements WritableComparable<PartGraph>{
	int joinNum;
	Map<QEdge, QEdge> graph;
	Set<QEdge> boundry = new HashSet<QEdge>();

	public PartGraph() {
		graph = new HashMap<SPGraphQuery.QEdge, SPGraphQuery.QEdge>();
		joinNum = 0x0;
	}
	public PartGraph(Map<QEdge, QEdge> graph, int num) {
		this.graph = new HashMap<SPGraphQuery.QEdge, SPGraphQuery.QEdge>(graph);
		joinNum = num;
	}
	
	@Override
	public void readFields(DataInput input) throws IOException {
		QEdge key = new QEdge();
		QEdge value = new QEdge();
		IntWritable jnum = new IntWritable();
		IntWritable size = new IntWritable();
		jnum.readFields(input);
		joinNum = jnum.get();
		size.readFields(input);
		for(int i = 0; i < size.get(); i++) {
			key.readFields(input);
			value.readFields(input);
			graph.put(key, value);
		}
	}
	@Override
	public void write(DataOutput output) throws IOException {
		new IntWritable(joinNum).write(output);
		new IntWritable(graph.entrySet().size()).write(output);
		for(Entry<QEdge, QEdge> entry : graph.entrySet()) {
			entry.getKey().write(output);
			entry.getValue().write(output);
		}
	}
	@Override
	public int compareTo(PartGraph other) {
		return joinNum > other.joinNum ? 1 : -1;
	}
	
	@Override
	public int hashCode() {
		return graph.hashCode() * 13 + joinNum;
	}
	@Override
	public boolean equals(Object obj) {
		PartGraph other = (PartGraph) obj;
		return this.graph.equals(other.graph) && this.joinNum == other.joinNum;
	}
	@Override
	public String toString() {
		return graph.toString();
	}
	
}
